//Copyright (c) Microsoft Corporation.  All rights reserved.

using System;
using System.Collections.Generic;
using System.Text;
using Eas.WebCrawler.Interfaces;
using Eas.WebCrawler.Model.Core;
using Eas.WebCrawler.Model.Actions;

namespace Eas.WebCrawler.Model.StateTrackers {
    internal class BaseStateTracker : Eas.WebCrawler.Common.BaseDisposable, StateTracker {
        int _WorkerCount;
        Queue<Eas.WebCrawler.Interfaces.WorkerState>[] _ChangeRequests;
        int[] _ActionCounts;
        WorkerContext _Context;
        static Eas.WebCrawler.Common.Threading.SpinWaitLock _Lock = new Eas.WebCrawler.Common.Threading.SpinWaitLock();
        BaseProcessingState[] _States;
        Eas.WebCrawler.Common.Threading.AsyncResultTracker _ResultTracker = new Eas.WebCrawler.Common.Threading.AsyncResultTracker();

        #region StateTracker Members

        protected override void Dispose(bool isDisposing) {
            try {
                if (isDisposing) {
                    foreach (BaseProcessingState state in _States) {
                        state.Dispose();
                    }
                }
            }
            finally {
                base.Dispose(isDisposing);
            }
        }

        public void StartAllWorkers() {
            for (int i = 0; i < _States.Length; i++) {
                BaseProcessingState.BeginNextAction(_States[i], _Context);
            }
        }

        public int GetCountOfWorkersInAction(ActionId actionId) {
            int count = GetCount(actionId);
            return count;
        }

        public void ReportStateChange(ProcessingState state, Eas.WebCrawler.Model.Actions.ProcessingAction oldAction, Eas.WebCrawler.Model.Actions.ProcessingAction newAction) {
            _Lock.Enter();
                try {
                    System.Diagnostics.Debug.Assert(oldAction.ActionId != newAction.ActionId);

                    int oldCount = System.Threading.Interlocked.Decrement(ref _ActionCounts[(int)oldAction.ActionId]);
                    int newCount = System.Threading.Interlocked.Increment(ref _ActionCounts[(int)newAction.ActionId]);
                    _Context.EventReporter.ReportTrace(TraceLevel.Annoying, state, "Action {0} set to count {1}.", oldAction.ActionId, oldCount);
                    _Context.EventReporter.ReportTrace(TraceLevel.Annoying, state, "Action {0} set to count {1}.", newAction.ActionId, newCount);
            
                    CheckStateCounts();
                    switch (state.WorkerState) {
                        case WorkerState.Paused:
                            if (GetCount(ActionId.Paused) == _WorkerCount) {
                                _Context.EventReporter.ReportTrace(TraceLevel.Annoying, state, "All workers are paused, reporting this fact.");
                                _Context.EventReporter.ReportPaused();
                            }
                            break;
                        case WorkerState.Stopped:
                            if (GetCount(ActionId.Terminal) == _WorkerCount) {
                                _Context.EventReporter.ReportTrace(TraceLevel.Annoying, state, "All workers are stopped, reporting this fact.");
                                _Context.EventReporter.ReportStopped();
                            }
                            break;
                        default:
                            break;
                    }
                }
                finally {
                    _Lock.Exit();
                }
        }

        private void CheckStateCounts() {
            int count = 0;
            for (int i = 0; i < _ActionCounts.Length; i++) {
                System.Diagnostics.Debug.Assert(_ActionCounts[i] >= 0);
                System.Diagnostics.Debug.Assert(_ActionCounts[i] <= _WorkerCount);
                count += _ActionCounts[i];
            }
            if (count != _WorkerCount) {
                System.Diagnostics.Debug.Assert(count == _WorkerCount);
                throw new System.ArgumentException(string.Format(System.Globalization.CultureInfo.InvariantCulture, "Worker Count {0} != {1}", count, _WorkerCount));
            }
            if(GetCount(ActionId.Terminal)>0 && (GetCount(ActionId.ReadStream)>0 || GetCount(ActionId.GetResponse)>0 )  ){
                TraceStateCounts(TraceLevel.Critical);
                _Context.EventReporter.ReportTrace(TraceLevel.Critical, BaseProcessingState.EmptyState, "A worker stalled early.");
                throw new System.ArgumentException(string.Format(System.Globalization.CultureInfo.InvariantCulture, "A worker stalled early.  TerminalCount:{0}\tGetResponseCount:{1}\tReadStreamCount:{2}", GetCount(ActionId.Terminal), GetCount(ActionId.GetResponse), GetCount(ActionId.ReadStream)));
            }
        }

        public void LoadChangeRequestsIntoState(ProcessingState state) {
            Queue<WorkerState> queue = _ChangeRequests[state.Id];
            if (queue.Count > 0) {
                lock (queue) {
                    while (queue.Count > 0) {
                        state.RequestChange(queue.Dequeue());
                    }
                }
            }
        }

        public BaseProcessingState this[int id] {
            get { return this._States[id]; }
        }

        public void OpenResult(IAsyncResult result, Eas.WebCrawler.Model.Actions.ProcessingAction action) {
            _ResultTracker.OpenResult(result, action.GetType().FullName);
        }

        public void CloseResult(IAsyncResult result) {
            _ResultTracker.CloseResult(result);
        }

        public void ValidateNoResultsAreOpened() {
            _ResultTracker.ValidateNoResultsAreOpened();
        }

        #endregion

        #region WorkerComponent Members

        public void Initialize(WorkerContext context) {
            _Context = context;
            _WorkerCount = _Context.WebCrawlerContext.WorkerCount;
            _ActionCounts = new int[Enum.GetValues(typeof(ActionId)).Length];
            _ActionCounts[(int)ActionId.Unstarted] = _WorkerCount;
            _ChangeRequests = new Queue<WorkerState>[_WorkerCount];
            _States = new BaseProcessingState[_WorkerCount];
            _ResultTracker.Clear();
            _Context.Events.Stopped += new EventHandler(Events_AfterStop);

            for (int i = 0; i < _WorkerCount; i++) {
                _ChangeRequests[i] = new Queue<WorkerState>();
                _States[i] = new BaseProcessingState(context, i);
            }
        }

        void Events_AfterStop(object sender, EventArgs e) {
            ValidateNoResultsAreOpened();
        }

        public void TraceStateCounts() {
            TraceLevel level = TraceLevel.Detailed;
            TraceStateCounts(level);
        }

        public void TraceStateCounts(TraceLevel level) {
            _Context.EventReporter.ReportTrace(level, BaseProcessingState.EmptyState, "ActionCounts:\t{0}:{1}\t{2}:{3}\t{4}:{5}\t{6}:{7}\t{8}:{9}\t{10}:{11}\t{12}:{13}",
                ActionId.Unstarted, _ActionCounts[(int)ActionId.Unstarted],
                ActionId.GetRequest, _ActionCounts[(int)ActionId.GetRequest],
                ActionId.GetResponse, _ActionCounts[(int)ActionId.GetResponse],
                ActionId.ReadStream, _ActionCounts[(int)ActionId.ReadStream],
                ActionId.PendingGetRequest, _ActionCounts[(int)ActionId.PendingGetRequest],
                ActionId.Terminal, _ActionCounts[(int)ActionId.Terminal],
                ActionId.Paused, _ActionCounts[(int)ActionId.Paused]);
        }
        #endregion

        private int GetCount(ActionId action) {
            return _ActionCounts[(int)action];
        }

        public void RequestChange(WorkerState changeRequest) {
            _Context.EventReporter.ReportTrace(TraceLevel.Detailed, BaseProcessingState.EmptyState, "Change '{0}' requested.", changeRequest);
            for (int i = 0; i < _ChangeRequests.Length; i++) {
                lock (_ChangeRequests[i]) {
                    _ChangeRequests[i].Enqueue(changeRequest);
                }
            }

            if (changeRequest == WorkerState.Stopped) {
                _Context.EventReporter.ReportTrace(TraceLevel.Detailed, BaseProcessingState.EmptyState, "Stop requested, aborting requests.");
                for (int i = 0; i < _States.Length; ++i) {
                    try { _States[i].AbortRequest(); }catch (System.Exception) { }
                }
            }
        }
    }
}
